package com.atwisdom.star.redis.util;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * spring redis 工具类
 * 
 **/
@SuppressWarnings(value = {"unchecked", "rawtypes"})
@Component
@Slf4j
public class RedisUtil {
    @Resource
    private RedisTemplate redisTemplate;

    /**
     * key释放存在
     *
     * @param key
     * @return
     */
    public boolean hasKey(String key) {
        return redisTemplate.hasKey(key);
    }

    /**
     * 缓存基本的对象，Integer、String、实体类等
     *
     * @param key   缓存的键值
     * @param value 缓存的值
     */
    public <T> void setCacheObject(final String key, final T value) {
        redisTemplate.opsForValue().set(key, value);
    }

    /**
     * 缓存基本的对象，Integer、String、实体类等
     *
     * @param key      缓存的键值
     * @param value    缓存的值
     * @param timeout  时间
     * @param timeUnit 时间颗粒度
     */
    public <T> void setCacheObject(final String key, final T value, final Integer timeout, final TimeUnit timeUnit) {
        redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
    }

    /**
     * 设置有效时间
     *
     * @param key     Redis键
     * @param timeout 超时时间
     * @return true=设置成功；false=设置失败
     */
    public boolean expire(final String key, final long timeout) {
        return expire(key, timeout, TimeUnit.SECONDS);
    }

    /**
     * 设置有效时间
     *
     * @param key     Redis键
     * @param timeout 超时时间
     * @param unit    时间单位
     * @return true=设置成功；false=设置失败
     */
    public boolean expire(final String key, final long timeout, final TimeUnit unit) {
        return redisTemplate.expire(key, timeout, unit);
    }

    /**
     * 获得缓存的基本对象。
     *
     * @param key 缓存键值
     * @return 缓存键值对应的数据
     */
    public <T> T getCacheObject(final String key) {
        ValueOperations<String, T> operation = redisTemplate.opsForValue();
        return operation.get(key);
    }

    public <T> T getPutCacheObject(final String key, Integer expireSeconds, Supplier supplier) {
        if (hasKey(key)) {
            return getCacheObject(key);
        }
        if (supplier != null) {
            T t = (T) supplier.get();
            if (t != null) {
                setCacheObject(key, t, expireSeconds, TimeUnit.MILLISECONDS);
                return t;
            }

        }
        return null;
    }

    /**
     * 删除单个对象
     *
     * @param key
     */
    public boolean deleteObject(final String key) {
        return redisTemplate.delete(key);
    }

    /**
     * 删除集合对象
     *
     * @param collection 多个对象
     * @return
     */
    public long deleteObject(final Collection collection) {
        return redisTemplate.delete(collection);
    }

    /**
     * 缓存List数据
     *
     * @param key      缓存的键值
     * @param dataList 待缓存的List数据
     * @return 缓存的对象
     */
    public <T> long setCacheList(final String key, final List<T> dataList) {
        Long count = redisTemplate.opsForList().rightPushAll(key, dataList);
        return count == null ? 0 : count;
    }

    /**
     * 获得缓存的list对象
     *
     * @param key 缓存的键值
     * @return 缓存键值对应的数据
     */
    public <T> List<T> getCacheList(final String key) {
        return redisTemplate.opsForList().range(key, 0, -1);
    }

    /**
     * 缓存Set
     *
     * @param key     缓存键值
     * @param dataSet 缓存的数据
     * @return 缓存数据的对象
     */
    public <T> long setCacheSet(final String key, final Set<T> dataSet) {
        Long count = redisTemplate.opsForSet().add(key, dataSet);
        return count == null ? 0 : count;
    }

    /**
     * 删除set中的元素
     *
     * @param key
     * @param values
     * @return
     */
    public Long removeCacheSet(final String key, Object... values) {
        Long count = redisTemplate.opsForSet().remove(key, values);
        return count;
    }

    /**
     * set中添加元素
     *
     * @param key
     * @param values
     * @return
     */
    public Long addCacheSet(final String key, Object... values) {
        Long count = redisTemplate.opsForSet().add(key, values);
        return count;
    }

    public Long deleteCacheSet(final String key, Object... values) {
        return redisTemplate.opsForSet().remove(key, values);
    }

    /**
     * 获得缓存的set
     *
     * @param key
     * @return
     */
    public <T> Set<T> getCacheSet(final String key) {
        return redisTemplate.opsForSet().members(key);
    }

    /**
     * 缓存Map
     *
     * @param key
     * @param dataMap
     */
    public <T> void setCacheMap(final String key, final Map<String, T> dataMap) {
        setCacheMap(key, dataMap, null);
    }

    public <T> void setCacheMap(final String key, final Map<String, T> dataMap, Long expireSeconds) {
        if (dataMap != null) {
            redisTemplate.opsForHash().putAll(key, dataMap);
            if (expireSeconds != null) {
                redisTemplate.expire(key, expireSeconds, TimeUnit.SECONDS);
            }
        }
    }

    public <T> Map<String, T> getPutCacheMap(final String key,
                                             Long expireSeconds, Supplier supplier) {
        if (hasKey(key)) {
            return getCacheMap(key);
        }
        Map<String, T> map = (Map<String, T>) supplier.get();
        if (map != null) {
            setCacheMap(key, map, expireSeconds);
        }
        return map;
    }

    public <T> T getPutCache(final String key, final Long timeout, final TimeUnit timeUnit, Supplier supplier) {
        if (redisTemplate.hasKey(key)) {
            return get(key);
        }
        T result = (T) supplier.get();
        if (result != null) {
            set(key, result, timeout, timeUnit);
        }
        return result;
    }

    public <T> T get(final String key) {
        ValueOperations<String, T> operation = redisTemplate.opsForValue();
        return operation.get(key);
    }

    public <T> void set(final String key, final T value, final Long timeout, final TimeUnit timeUnit) {
        redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
    }

    /**
     * 获得缓存的Map
     *
     * @param key
     * @return
     */
    public <T> Map<String, T> getCacheMap(final String key) {
        return redisTemplate.opsForHash().entries(key);
    }

    /**
     * 往Hash中存入数据
     *
     * @param key   Redis键
     * @param hKey  Hash键
     * @param value 值
     */
    public <T> void setCacheMapValue(final String key, final String hKey, final T value) {
        redisTemplate.opsForHash().put(key, hKey, value);
    }

    /**
     * 获取Hash中的数据
     *
     * @param key  Redis键
     * @param hKey Hash键
     * @return Hash中的对象
     */
    public <T> T getCacheMapValue(final String key, final String hKey) {
        HashOperations<String, String, T> opsForHash = redisTemplate.opsForHash();
        return opsForHash.get(key, hKey);
    }

    /**
     * 获取多个Hash中的数据
     *
     * @param key   Redis键
     * @param hKeys Hash键集合
     * @return Hash对象集合
     */
    public <T> List<T> getMultiCacheMapValue(final String key, final Collection<Object> hKeys) {
        return redisTemplate.opsForHash().multiGet(key, hKeys);
    }

    /**
     * 获得缓存的基本对象列表
     *
     * @param pattern 字符串前缀
     * @return 对象列表
     */
    public Collection<String> keys(final String pattern) {
        return redisTemplate.keys(pattern);
    }


    public Boolean setBit(final String key, Long offest, boolean tag) {
        return redisTemplate.opsForValue().setBit(key, offest, tag);
    }

    /**
     * @param key
     * @param offest
     * @return
     */
    public Boolean getBit(final String key, Long offest) {
        return redisTemplate.opsForValue().getBit(key, offest);
    }

    /**
     * 递增
     *
     * @param key
     * @return
     */
    public Long increment(final String key) {
        return redisTemplate.opsForValue().increment(key);
    }

    public Long increment(final String key, Long v) {
        return redisTemplate.opsForValue().increment(key, v);
    }

    /**
     * 递减
     *
     * @param key
     * @return
     */
    public Long decrement(final String key) {
        return redisTemplate.opsForValue().decrement(key);
    }

    public Long decrement(final String key, Long v) {
        return redisTemplate.opsForValue().decrement(key, v);
    }

    /**
     * 基数统计
     *
     * @return
     */
    public HyperLogLogOperations hyperLogLogOperations() {
        return redisTemplate.opsForHyperLogLog();
    }

    /**
     * 地理位置
     *
     * @return
     */
    public GeoOperations geoOperations() {
        return redisTemplate.opsForGeo();
    }

    /**
     * 查询多个key下的值
     * @param keyList key 列表
     * @param <T> 返回值类型
     * @return list
     */
    public <T> List<T> getCache(Set<String> keyList) {
        return (List<T>) redisTemplate.opsForValue().multiGet(keyList);
    }


    /**
     * 创建group
     * @param streamName
     * @param groupName
     * @param readOffset
     * @return
     */
    public String createGroup(String streamName, String groupName, ReadOffset readOffset) {
        String group = null;
        Long size = redisTemplate.opsForStream().size(streamName);
        if (size == null || size < 1) {
            group = redisTemplate.opsForStream().createGroup(streamName, readOffset, groupName);
        } else {
            group = redisTemplate.opsForStream().createGroup(streamName, groupName);
        }
        return group;
    }

    /**
     * 发消息
     * @param streamName
     * @param messageJson
     * @return
     */
    public  <T> RecordId addMapMessage(String streamName, T messageJson) {
        ObjectRecord<String, T> record = StreamRecords.newRecord()
                .in(streamName)
                .ofObject(messageJson)
                .withId(RecordId.autoGenerate());

        RecordId recordId = redisTemplate.opsForStream().add(record);
        log.info("返回的record-id: {}", recordId);
        return recordId;
    }

    /**
     * 读取一条消息
     * @param streamName
     * @param groupName
     * @param consumerName
     * @param duration
     * @param readOffset
     * @return
     */
    public MapRecord<String, Object, Object> readOne(String streamName, String groupName, String consumerName, Duration duration, ReadOffset readOffset) {
        List<MapRecord<String, Object, Object>> recordList = redisTemplate.opsForStream().read(
                Consumer.from(groupName, consumerName),
                StreamReadOptions.empty().count(1).block(duration),
                StreamOffset.create(streamName, readOffset)
        );
        if (recordList == null || recordList.isEmpty()) {
            return null;
        }
        return recordList.get(0);

    }

    /**
     * 确认消息
     * @param streamName
     * @param groupName
     * @param recordId
     * @return
     */
    public Long ack(String streamName,String groupName, RecordId recordId) {
        return redisTemplate.opsForStream().acknowledge(streamName, groupName, recordId);
    }

    /**
     * 删除消息
     * @param streamName
     * @param recordId
     * @return
     */
    public Long delete(String streamName,RecordId recordId) {
        return redisTemplate.opsForStream().delete(streamName, recordId);
    }
}
